package com.example.kafkademo.multithread;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created by zyy on 2021/12/29.
 */
public class WorkConsumer {

    public static void main(String[] args) {
        Properties props = new Properties();  //***
        props.put("bootstrap.servers", "10.42.25.12:9092,10.42.25.13:9092,10.42.25.15:9092");
        props.put("group.id", "12334");
        props.put("enable.auto.commit", "true");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        int workerNum = 5;

        ExecutorService executors = new ThreadPoolExecutor(workerNum, workerNum, 0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(1000),new ThreadPoolExecutor.CallerRunsPolicy());
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Arrays.asList("test-a"));

        while (true)  {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));

//            for (ConsumerRecord record : records) {
//                System.out.println(record);
                executors.submit(new Worker(records));
            System.out.println(records);
//            }
        }
    }
}

class Worker extends Thread{
    public final ConsumerRecords<String,String> records;
    public Worker(ConsumerRecords<String,String> record){
        this.records = record;
    }

    @Override
    public void run() {
        try {
            Thread.sleep(100);
//            System.out.printf("offset = %d, value = %s, parition = %d", record.offset(), record.value(),record.partition());
//            System.out.println(Thread.currentThread().getName() + "=====:" + record.topic() + ": " + record.partition() );
//            record.partition();  //处理逻辑
            System.out.println(Thread.currentThread().getName() +"   count:" + records.count());
            System.out.println(records);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("***************");
        System.out.println(Thread.currentThread().getName());
        //处理records
    }
}